package hbase;


import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.sql.SQLOutput;
import java.util.ArrayList;
import java.util.List;

import static org.apache.zookeeper.ZooDefs.OpCode.delete;

/**
 * 连接Hbase 首先要知道zookeeper部署在哪里
 *
 * 1.初始化连接
 * 2.创建操作对象
 *      ddl操作：需要Admin对象
 *      dml操作：Table
 * 3.操作数据
 *      ddl操作
 *          建表 create 'myUser','f1','f2'
 *          删表 disable 'myUser'
 *               drop 'myUser'
 *          修改表 alter 'myUser',NAME=>'f1',VERSIONS=>3
 *      dml操作
 *          新增、修改 put 'myUser','rk001','f1:name','tom'
 *          删除 delete 'myUser','rk001','f1:name'
 *          查询
 *              get 'myUser','rk001','f1','f2:phone'
 *              //scan 'myUser',{COLUMNS=>['f1','f2:phone']}
 *              scan 'myUser',{STARTROW=>'0003',ENDROW=>'0006'}
 *              过滤器查询
 *                  ...
 *                  比较过滤器 CompareFilter
 *                  分页过滤器 PageFilter
 *                  ...
 *
 * 4.关闭
 */
public class hbaseOperation {

    public Connection connection;

    public final String TABLE_NAME="myUser";
    public final String TABLE_NAME_DELETE="user";

    /**
     * 初始化连接
     */
    @Before
    public void init() throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum","master:2181,slave1:2181,slave2:2181");
        connection = ConnectionFactory.createConnection(config);
    }

    /**
     * 关闭连接
     */
    @After
    public void close() throws IOException {
        if(connection != null){
            connection.close();
        }
    }

    /**
     * 建表 create 'myUser','f1','f2'
     * Java中一切皆对象
     */
    @Test
    public void create() throws IOException {
        //1.获取操作对象
        Admin admin = connection.getAdmin();
        //2.构建表描述对象
        TableDescriptorBuilder tdb = TableDescriptorBuilder.newBuilder(TableName.valueOf(TABLE_NAME));

        //3.构建列族(描述对象)
        ColumnFamilyDescriptor f1 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("f1")).build();
        ColumnFamilyDescriptor f2 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("f2")).build();
        tdb.setColumnFamilies(Lists.newArrayList(f1,f2));//和表的描述对象进行关联
        //4.建表
        admin.createTable(tdb.build());
        admin.close();
    }

    /**
     * 删表 disable 'user'
     *      drop 'user'
     */
    @Test
    public void deleteTable() throws IOException {
        //1.获取操作对象
        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf(TABLE_NAME_DELETE);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
        admin.close();
    }

    /**
     * 修改表 alter 'myUser',NAME=>'f1',VERSIONS=>3
     */
    @Test
    public void alterTable() throws IOException {
        //1.获取操作对象
        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf(TABLE_NAME);
        ColumnFamilyDescriptor f1 =
                ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("f1")).setMaxVersions(3).build();
        admin.modifyColumnFamily(tableName,f1);
        admin.close();
    }

    /**
     *  DML : 新增、修改 put 'myUser','rk001','f1:name','tom'
     */
    @Test
    public void addData() throws IOException {
        //1.构建table对象
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //构建put对象
        Put put = new Put(Bytes.toBytes("0001"));
        //put 一行数据
        put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("name"),Bytes.toBytes("Tom"));
        put.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(18));
        put.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(124423));
        put.addColumn("f1".getBytes(),"address".getBytes(),Bytes.toBytes("上海市"));
        table.put(put);
        table.close();
    }

    /**
     * hbase的批量插入数据
     */
    @Test
    public void batchInsert() throws IOException {
        //1.构建table对象
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        // 创建put对象，并指定rowkey
        Put put = new Put("0002".getBytes());

        put.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(1));
        put.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("曹操"));
        put.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(30));
        put.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("沛国谯县"));
        put.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("16888888888"));
        put.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("helloworld"));

        Put put2 = new Put("0003".getBytes());
        put2.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(2));
        put2.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("刘备"));
        put2.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(32));
        put2.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put2.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("幽州涿郡涿县"));
        put2.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("17888888888"));
        put2.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("talk is cheap , show me the code"));

        Put put3 = new Put("0004".getBytes());
        put3.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(3));
        put3.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("孙权"));
        put3.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(35));
        put3.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put3.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("下邳"));
        put3.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("12888888888"));
        put3.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("what are you 弄啥嘞！"));

        Put put4 = new Put("0005".getBytes());
        put4.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(4));
        put4.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("诸葛亮"));
        put4.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(28));
        put4.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put4.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("四川隆中"));
        put4.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("14888888888"));
        put4.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("出师表你背了嘛"));

        Put put5 = new Put("0006".getBytes());
        put5.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(5));
        put5.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("司马懿"));
        put5.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(27));
        put5.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put5.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("哪里人有待考究"));
        put5.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("15888888888"));
        put5.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("跟诸葛亮死掐"));


        Put put6 = new Put("0007".getBytes());
        put6.addColumn("f1".getBytes(),"id".getBytes(),Bytes.toBytes(5));
        put6.addColumn("f1".getBytes(),"name".getBytes(),Bytes.toBytes("xiaobubu—吕布"));
        put6.addColumn("f1".getBytes(),"age".getBytes(),Bytes.toBytes(28));
        put6.addColumn("f2".getBytes(),"sex".getBytes(),Bytes.toBytes("1"));
        put6.addColumn("f2".getBytes(),"address".getBytes(),Bytes.toBytes("内蒙人"));
        put6.addColumn("f2".getBytes(),"phone".getBytes(),Bytes.toBytes("15788888888"));
        put6.addColumn("f2".getBytes(),"say".getBytes(),Bytes.toBytes("貂蝉去哪了"));

        List<Put> listPut = new ArrayList<Put>();
        listPut.add(put);
        listPut.add(put2);
        listPut.add(put3);
        listPut.add(put4);
        listPut.add(put5);
        listPut.add(put6);

        table.put(listPut);
        table.close();
    }

    /**
     *  删除 delete 'myUser','rk001','f1:name'
     */
    @Test
    public void deleteData() throws IOException {
        //1.构建table对象
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //构建删除对象
        Delete delete = new Delete(Bytes.toBytes("0001"));

        delete.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("name"));
        table.delete(delete);
        table.close();
    }

    /**
     *  get方式查询 get 'myUser','rk001','f1','f2:phone'
     *  Table --> Result对象
     */
    @Test
    public void getData() throws IOException {
        //1.构建table对象
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //构建get对象
        Get get = new Get(Bytes.toBytes("0001"));

        get.addFamily(Bytes.toBytes("f1"));
        get.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("name"));
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        //list集合在使用的过程中 小心空指针异常
        if(cells == null){
            System.out.println("============空数据===========");
        } else {
            cells.forEach( cell -> {
                //获取列族名
                byte[] familyName = CellUtil.cloneFamily(cell);
                // 获取列名
                byte[] columnName = CellUtil.cloneQualifier(cell);
                // 获取rowKey
                byte[] rowKey = CellUtil.cloneRow(cell);
                // 获取cell值
                byte[] cellValue = CellUtil.cloneValue(cell);
                // 需要判断字段的数据类型，使用对应的转换的方法，才能够获取到值
                // 拿到columnName为id和age的 数据
                if ("age".equals(Bytes.toString(columnName)) || "id".equals(Bytes.toString(columnName))) {  //why?
                    System.out.println(Bytes.toString(familyName));
                    System.out.println(Bytes.toString(columnName));
                    System.out.println(Bytes.toString(rowKey));
                    System.out.println(Bytes.toInt(cellValue));
                } else {
                    System.out.println(Bytes.toString(familyName));
                    System.out.println(Bytes.toString(columnName));
                    System.out.println(Bytes.toString(rowKey));
                    System.out.println(Bytes.toString(cellValue));
                }
            });
        }
        table.close();
    }

    /**
     * scan方式查询：scan 'myUser',{STARTROW=>'0003',ENDROW=>'0006'}
     */
    @Test
    public void scandata() throws IOException {
        //1.构建table对象
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //构建scan对象
        Scan scan = new Scan()
                .withStartRow(Bytes.toBytes("0003"))  //默认包含0003
                .withStopRow(Bytes.toBytes("0006"))   //默认不包含0006
                .addFamily(Bytes.toBytes("f1"))
                .addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"));
        ResultScanner scanner = table.getScanner(scan);
        //与getData（）重复，可单独抽出一个方法
        scanner.forEach(iter -> {
            List<Cell> cells = iter.listCells();
            //list集合在使用的过程中 小心空指针异常
            if(cells == null){
                System.out.println("============空数据===========");
            } else {
                cells.forEach( cell -> {
                    //获取列族名
                    byte[] familyName = CellUtil.cloneFamily(cell);
                    // 获取列名
                    byte[] columnName = CellUtil.cloneQualifier(cell);
                    // 获取rowKey
                    byte[] rowKey = CellUtil.cloneRow(cell);
                    // 获取cell值
                    byte[] cellValue = CellUtil.cloneValue(cell);
                    // 需要判断字段的数据类型，使用对应的转换的方法，才能够获取到值
                    // 拿到columnName为id和age的 数据
                    if ("age".equals(Bytes.toString(columnName)) || "id".equals(Bytes.toString(columnName))) {  //why?
                        System.out.println(Bytes.toString(familyName));
                        System.out.println(Bytes.toString(columnName));
                        System.out.println(Bytes.toString(rowKey));
                        System.out.println(Bytes.toInt(cellValue));
                    } else {
                        System.out.println(Bytes.toString(familyName));
                        System.out.println(Bytes.toString(columnName));
                        System.out.println(Bytes.toString(rowKey));
                        System.out.println(Bytes.toString(cellValue));
                    }
                });
            }
        });
        //与getData（）重复，可单独抽出一个方法

        table.close();
    }

    /**
     *  过滤器--比较过滤器
     *  cell值 传进来 --》 比较操作 根据什么比较？比较器
     *  比较操作: 小于 大于 等于 ...
     *  比较器有哪些 ByteArrayComparable
     *                  |-- LongComparator (org.apache.hadoop.hbase.filter)
     *                      BinaryComparator (org.apache.hadoop.hbase.filter) 二进制
     *                      RegexStringComparator (org.apache.hadoop.hbase.filter) 正则表达式
     *                      NullComparator (org.apache.hadoop.hbase.filter)
     *                      BigDecimalComparator (org.apache.hadoop.hbase.filter)
     *                      BitComparator (org.apache.hadoop.hbase.filter)  根据位运算
     *                      SubstringComparator (org.apache.hadoop.hbase.filter)
     *                          举例：scan 'user',{COLUMNS=>['info','data'],FILTER=>"FamilyFilter(=,'substring:a')",VERSION=>3}
     */

}
